package com.lagou.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author: yehw
 * @date: 2021/1/21 下午10:26
 * @description: 翻滚窗口
 */
public class TumblingWindow {


    public static void main(String[] args) throws Exception {
        //环境上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> linux121 = env.socketTextStream("linux121", 7777);
        SingleOutputStreamOperator<Tuple2<String, Long>> streamOperator = linux121.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                //当前时间
                long currentTimeMillis = System.currentTimeMillis();

                return new Tuple2<String, Long>(s, currentTimeMillis);
            }
        });
        KeyedStream<Tuple2<String, Long>, Tuple> tuple2TupleKeyedStream = streamOperator.keyBy(0);
        //基于时间驱动，每10秒划分一个窗口
        WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> tuple2TupleTimeWindowWindowedStream = tuple2TupleKeyedStream.timeWindow(Time.seconds(10));
        //tuple2TupleTimeWindowWindowedStream.apply(new MyTimeWindowFunction()).print();
        tuple2TupleTimeWindowWindowedStream.apply(new WindowFunction<Tuple2<String, Long>, Object, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<Object> out) throws Exception {
                SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                long start = window.getStart();
                long end = window.getEnd();
                AtomicLong sum = new AtomicLong();
                input.forEach(x -> {
                    Tuple2<String, Long> x1 = (Tuple2<String, Long>) x;
                    sum.addAndGet(x1.f1);
                });
                out.collect("key:" + tuple.getField(0) + " value: " + sum +
                        "| window_start :" + format.format(start) +
                        " window_end :" + format.format(end));
            }
        }).print();
        env.execute();
    }
}
